package io.ai.arch.java.mq.rabbitmq;

import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

@Slf4j
//@Configuration
public class RabbitMQConfig {

    public final static String exchangeName = "ai-arch-java-exchange";

    public final static String biz1 = "-simple";
    public final static String routingKey = "ai-arch-java-routingkey" + biz1;
    public static final String queueName = "ai-arch-java-queue" + biz1;
    public static final String groupName = "ai-arch-java-group" + biz1;

    public final static String biz2 = "-test";
    public static final String biz2RoutingKey = "ai-arch-java-routingkey" + biz2;
    public static final String biz2QueueName = "ai-arch-java-queue" + biz2;
    public static final String biz2groupName = "ai-arch-java-group" + biz2;

    public static final String highPriorityStr = "10";
    public static final Integer highPriority = Integer.valueOf(highPriorityStr);
    public static final String mediumPriorityStr = "5";
    public static final Integer mediumPriority = Integer.valueOf(mediumPriorityStr);
    public static final String lowPriorityStr = "1";
    public static final Integer lowPriority = Integer.valueOf(lowPriorityStr);

    @Bean
    public DirectExchange exchange() {
        return new DirectExchange(exchangeName);
    }

    @Bean
    public Queue queue() {
        Queue queue = new Queue(queueName);
//        queue.addArgument("x-message-ttl", 10000);
//        queue.addArgument("x-max-priority", highPriority);
        return queue;
    }

    @Bean
    public Binding binding() {
        return BindingBuilder.bind(queue()).to(exchange()).with(routingKey);
    }

    @Bean
    public Queue biz2Queue() {
        Queue queue = new Queue(biz2QueueName);
        queue.addArgument("x-dead-letter-exchange", exchangeName);
        queue.addArgument("x-dead-letter-routing-key", routingKey);
//        queue.addArgument("x-message-ttl", 10000);
//        queue.addArgument("x-max-priority", highPriority);
        return queue;
    }

    @Bean
    public Binding biz2Binding() {
        return BindingBuilder.bind(biz2Queue()).to(exchange()).with(biz2RoutingKey);
    }

    //    @Bean
    public Queue asyncExportQueue(@Qualifier("rabbitAdmin") RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue("asyncExportQueue", true, false, false, null);
        Binding binding = new Binding("asyncExportQueue", Binding.DestinationType.QUEUE, "asyncExportExchange", "asyncExportQueue", null);
        rabbitAdmin.declareQueue(queue);
        rabbitAdmin.declareBinding(binding);
        return queue;
    }

    /**
     * 异步导出交换机
     * @param rabbitAdmin
     * @return
     */
//    @Bean
    public DirectExchange asyncExportExchange(@Qualifier("rabbitAdmin") RabbitAdmin rabbitAdmin) {
        //声明交换机
        DirectExchange directExchange = new DirectExchange("asyncExportExchange", true, false, null);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }


//    static CountDownLatch consumerCountDownLatch = new CountDownLatch(1);

    //    @RabbitListener(queues = queueName, group = groupName, priority = highPriorityStr/*, concurrency = "2"*/)
    public void listen10(Message message) {
//        try {
//            consumerCountDownLatch.await();
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
        priorityCountMap.computeIfAbsent(AiMessagePriority.HIGH.priority, (a) -> new AtomicLong(0)).incrementAndGet();
//        log.info("listen10: {}", message);
    }

    //    @RabbitListener(queues = queueName, group = groupName, priority = mediumPriorityStr)
    public void listen20(Message message) {
//        try {
//            consumerCountDownLatch.await();
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
        priorityCountMap.computeIfAbsent(AiMessagePriority.MEDIUM.priority, (a) -> new AtomicLong(0)).incrementAndGet();
//        log.info("listen20: {}", message);
    }

    //    @RabbitListener(queues = queueName, group = groupName, priority = lowPriorityStr)
    public void listen30(Message message) {
//        try {
//            consumerCountDownLatch.await();
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
        priorityCountMap.computeIfAbsent(AiMessagePriority.LOW.priority, (a) -> new AtomicLong(0)).incrementAndGet();
//        log.info("listen30: {}", message);
    }

    @RabbitListener(queues = queueName/*, concurrency = "2"*/)
    public void listenMessage(Message message) {
//        try {
//            consumerCountDownLatch.await();
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
//        priorityCountMap.computeIfAbsent(message.getMessageProperties().getPriority(), (a) -> new AtomicLong(0)).incrementAndGet();
        log.info("消费中 {}", message.getMessageProperties().getExpiration());
//        log.info("listenMessage: {}", message);
//        log.info("listenMessage: {}", channel);
    }

    //    @Bean
    public RabbitProducer rabbitProducer() {
        return new RabbitProducer();
    }

    //    @Scheduled(fixedDelay = 1000)
    public void printPriorityCount() {
//        try {
//            consumerCountDownLatch.await();
//        } catch (InterruptedException e) {
//            throw new RuntimeException(e);
//        }
        log.info("priorityCountMap： {}", priorityCountMap);
    }


    static Map<Integer, AtomicLong> priorityCountMap = new HashMap<>();


    static class RabbitProducer implements ApplicationRunner {

        @Resource
        public RabbitTemplate rabbitTemplate;

        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

        AtomicLong count = new AtomicLong(0);

        @Override
        public void run(ApplicationArguments args) {
//            IntStream.range(0, 5).forEach(i -> sendBiz2Message());
//            IntStream.range(1, 5).forEach(i -> sendAllPriorityMessage());
//            while (true) sendAllPriorityMessage();

            sendBiz2MessageOffSet(Duration.ofSeconds(10));
            sendBiz2MessageOffSet(Duration.ofMinutes(1));
//            sendBiz2Message(LocalDateTime.parse("2025-06-08 12:13:00", dateTimeFormatter));
//            sendBiz2Message(LocalDateTime.parse("2025-06-08 12:14:12", dateTimeFormatter));
        }

        private void sendBiz2MessageOffSet(Duration duration) {
            log.info("{}", LocalDateTime.now().plus(duration));
            Message ttlMessage = MessageBuilder.withBody("message".getBytes()).setExpiration(String.valueOf(duration.toMillis())).build();
            rabbitTemplate.send(exchangeName, biz2RoutingKey, ttlMessage, new CorrelationData());
        }

        private void sendBiz2Message(LocalDateTime runAt) {
            // runAt和now的差异的毫秒数
            long ttl = Duration.between(LocalDateTime.now(), runAt).toMillis();
            Message ttlMessage = MessageBuilder.withBody("message".getBytes()).setExpiration(String.valueOf(ttl)).build();
            rabbitTemplate.send(exchangeName, biz2RoutingKey, ttlMessage, new CorrelationData());

//            ttlMessage = MessageBuilder.withBody("message".getBytes())
//                    .setExpiration("20000")
//                    .build();
//            rabbitTemplate.send(exchangeName, biz2RoutingKey, ttlMessage, new CorrelationData());
        }

        private void sendAllPriorityMessage() {
            Message highMessage = MessageBuilder.withBody("hello".getBytes()).setPriority(AiMessagePriority.HIGH.priority).build();
            rabbitTemplate.send(exchangeName, routingKey, highMessage, new CorrelationData());
            Message mediumMessage = MessageBuilder.withBody("hello".getBytes()).setPriority(AiMessagePriority.MEDIUM.priority).build();
            rabbitTemplate.send(exchangeName, routingKey, mediumMessage, new CorrelationData());
            Message lowMessage = MessageBuilder.withBody("hello".getBytes()).setPriority(AiMessagePriority.LOW.priority).build();
            rabbitTemplate.send(exchangeName, routingKey, lowMessage, new CorrelationData());
            count.incrementAndGet();
            if (count.get() % 10000 == 0) {
//                log.info("priorityCountMap： {}", priorityCountMap);
            }
        }

        public AiMessagePriority getRandomPriority() {
            int randomNum = (int) (Math.random() * AiMessagePriority.HIGH.priority);
            if (randomNum > AiMessagePriority.MEDIUM.priority) {
                return AiMessagePriority.HIGH;
            } else if (randomNum > AiMessagePriority.LOW.priority) {
                return AiMessagePriority.MEDIUM;
            } else {
                return AiMessagePriority.LOW;
            }
        }
    }


    enum AiMessagePriority {
        HIGH(highPriority), MEDIUM(mediumPriority), LOW(lowPriority);

        private Integer priority;

        AiMessagePriority(Integer priority) {
            this.priority = priority;
        }

        public static Integer getPriority(String priority) {
            switch (priority) {
                case "HIGH":
                    return highPriority;
                case "LOW":
                    return lowPriority;
                default:
                    return mediumPriority;
            }
        }
    }

    @Bean
    public RabbitAdmin rabbitAdmin(RabbitTemplate rabbitTemplate) {
        return new RabbitAdmin(rabbitTemplate);
    }


//    @Component
    @RequiredArgsConstructor
    public static class RabbiMQHelper implements ApplicationRunner {

        private final RabbitAdmin rabbitAdmin;
        @Value("${spring.application.name}")
        private String appName;

        @Override
        public void run(ApplicationArguments args) {

            String exchangeName = "exchange-" + appName.replaceAll(".", "-");
            DirectExchange directExchange = new DirectExchange(exchangeName);
            rabbitAdmin.declareExchange(directExchange);

            EnumSet.allOf(DELAY_LEVEL.class).forEach(delayLevel -> {
                String queueName = "queue-" + appName.replaceAll(".", "-") +  "-" + delayLevel.getDelaySeconds();
                Queue queue = new Queue(queueName);
                rabbitAdmin.declareQueue(queue);
                String routingKey = "key-" + delayLevel.getDelaySeconds();
                Binding binding = BindingBuilder.bind(queue).to(directExchange).with(routingKey);
                rabbitAdmin.declareBinding(binding);
            });
        }
    }


    /**
     * 支持1、5、10、30、45、60、2min、5min、10min、15min、30min、45min、60min、2h、4h、6h、8h、12h、24h、2d、3d的延迟级别
     */
    enum DELAY_LEVEL {
        ZERO_SECOND(0),  // 0秒延迟
        FIVE_SECOND(5),  // 10秒延迟
        TEN_SECOND(10),  // 10秒延迟
        THIRTY_SECOND(30),  // 30秒延迟
        ONE_MINUTE(1 * 60),        // 1分钟延迟
        FIVE_MINUTE(5 * 60),  // 5分钟延迟
        TEN_MINUTE(10 * 60), // 10分钟延迟
        THIRTY_MINUTE(30 * 60),  // 30分钟延迟
        ONE_HOUR(1 * 60 * 60), //  1小时延迟
        EIGHT_HOUR(2 * 60 * 60), // 2小时延迟
        ;
        private Integer delaySeconds;

        DELAY_LEVEL(Integer delaySeconds) {
            this.delaySeconds = delaySeconds;
        }

        public Integer getDelaySeconds() {
            return delaySeconds;
        }
    }
}
